package com.zhang.flink.window;

import com.zhang.flink.bean.ClickEvent;
import com.zhang.flink.bean.UrlViewCountWindow;
import com.zhang.flink.example.source.ClickSource;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

/**
 * 每隔5s输出过去10s的url访问情况
 *
 * @title: 使用KeyedProcessFunction实现滑动窗口
 * @author: zhang
 * @date: 2022/2/18 19:00
 */
public class SlidingWindowInKeyedProcess {
    public static void main(String[] args) throws Exception {
        //todo 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickSource())
                .keyBy(r -> r.url)
                .process(new SlidingWindowFunc())
                .print();

        //todo 执行任务
        env.execute();
    }

    public static class SlidingWindowFunc extends KeyedProcessFunction<String, ClickEvent, UrlViewCountWindow> {
        //定义滑动窗口长度
        private long slidingWindowSize = 10 * 1000L;
        //定义滑动窗口步长
        private long slidingWindowStep = 5 * 1000L;
        //key:窗口开始时间
        //value：属于窗口所有事件
        private MapState<Long, List<ClickEvent>> mapState;

        @Override
        public void open(Configuration parameters) throws Exception {
            mapState = getRuntimeContext().getMapState(
                    new MapStateDescriptor<Long, List<ClickEvent>>(
                            "map-state",
                            Types.LONG, Types.LIST(Types.POJO(ClickEvent.class))
                    )
            );
        }

        @Override
        public void processElement(ClickEvent value, Context ctx, Collector<UrlViewCountWindow> out) throws Exception {
            //计算窗口时间
            long currTs = ctx.timerService().currentProcessingTime();
            ArrayList<Long> arrayList = new ArrayList<>((int) (slidingWindowSize / slidingWindowStep));
            long lastStart = getWindowStart(currTs, slidingWindowStep);
            for (long start = lastStart; start > currTs - slidingWindowSize; start -= slidingWindowStep) {
                arrayList.add(start);
            }

            for (Long startWin : arrayList) {
                if (!mapState.contains(startWin)) {
                    ArrayList<ClickEvent> events = new ArrayList<>();
                    events.add(value);
                    mapState.put(startWin, events);
                } else {
                    mapState.get(startWin).add(value);
                }
                ctx.timerService().registerProcessingTimeTimer(startWin + slidingWindowSize - 1L);
            }


        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<UrlViewCountWindow> out) throws Exception {
            long winEnd = timestamp + 1L;
            long winStart = winEnd - slidingWindowSize;
            String url = ctx.getCurrentKey();
            long size = mapState.get(winStart).size();
            out.collect(new UrlViewCountWindow(
                    url,
                    size,
                    winStart,
                    winEnd
            ));
            mapState.remove(winStart);
        }
    }


    public static long getWindowStart(long currTs, long step) {
        return currTs - (currTs + step) % step;
    }
}
